package com.ferry.transport.netty.consumer;

import com.ferry.registry.domain.registry.NodeEvent;
import com.ferry.registry.domain.registry.RegistryMetaData;
import com.ferry.registry.listener.NotifyListener;
import com.ferry.transport.api.*;
import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.ferry.transport.netty.client.ConsumerClient;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by daiyong
 */
public class DefaultConnectionWatcher implements ConnectionWatcher {

	private ConnectionManager connectionManager;

	private Connector<Connection> connector;

	private ConsumerClient consumerClient;

	private final Lock lock = new ReentrantLock();

	private final Condition condition = lock.newCondition();

	private AtomicBoolean needSignal = new AtomicBoolean(false);

	private Directory directory;

	public DefaultConnectionWatcher(Directory directory) {
		this.directory = directory;
	}

	@Override
	public void start() {
		//订阅服务
		consumerClient.subscribe(directory, new NotifyListener() {
			@Override
			public void notify(RegistryMetaData registryMetaData, NodeEvent nodeEvent) {
				UnResolvedAddress address = new UnResolvedSocketAddress(registryMetaData.getServiceAddress().getIp(), registryMetaData.getServiceAddress().getPort());
				//创建channelGroup
				WrappedChannelGroup wrappedChannelGroup = connector.group(address);
				//新增节点
				if (nodeEvent == NodeEvent.LEAF_ADD) {
					//channle group 可用
					if (wrappedChannelGroup.isAvailable()) {
						onSucceed(wrappedChannelGroup, needSignal.getAndSet(false));
					} else {
						if (wrappedChannelGroup.isConnecting()) {
							wrappedChannelGroup.onAvailable(() -> onSucceed(wrappedChannelGroup, needSignal.getAndSet(false)));
						} else {
							wrappedChannelGroup.setConnecting(true);
							Connection[] connections = connectTo(address, wrappedChannelGroup, registryMetaData, true);
							final AtomicInteger countdown = new AtomicInteger(connections.length);
							for (Connection c : connections) {
								c.operationComplete(isSuccess -> {
									if (isSuccess) {
										onSucceed(wrappedChannelGroup, needSignal.getAndSet(false));
									}
									if (countdown.decrementAndGet() <= 0) {
										wrappedChannelGroup.setConnecting(false);
									}
								});
							}
						}
					}
				}
				//删除节点
				else {
					connector.removeChannelGroup(directory, wrappedChannelGroup);
					wrappedChannelGroup.removeWeight(directory);
				}
			}

			/**
			 * 链接到服务端
			 * @param address
			 * @param wrappedChannelGroup
			 * @param registryMetaData
			 * @param b
			 * @return
			 */
			private Connection[] connectTo(UnResolvedAddress address, WrappedChannelGroup wrappedChannelGroup, RegistryMetaData registryMetaData, boolean async) {
				int connCount = registryMetaData.getConnCount(); //链接数量

				Connection[] connections = new Connection[connCount];
				wrappedChannelGroup.setCapacity(connCount);

				for (int i = 0; i < connCount; i++) {
					Connection connection = connector.connect(address, async);
					connections[i] = connection;
					connectionManager.manage(connection);
				}

				consumerClient.offlineListening(address, () -> {
					connectionManager.cancelAutoReconnect(address);
					if (!wrappedChannelGroup.isAvailable()) {
						connector.removeChannelGroup(directory, wrappedChannelGroup);
					}
				});

				return connections;
			}

			private void onSucceed(WrappedChannelGroup wrappedChannelGroup, boolean doSignal) {
				connector.addChannelGroup(directory, wrappedChannelGroup);
				if (doSignal) {
					lock.lock();
					try {
						condition.signalAll();
					} finally {
						lock.unlock();
					}
				}
			}

		});
	}

	public void setConnectionManager(ConnectionManager connectionManager) {
		this.connectionManager = connectionManager;
	}

	public void setConnector(Connector<Connection> connector) {
		this.connector = connector;
	}

	public void setConsumerClient(ConsumerClient consumerClient) {
		this.consumerClient = consumerClient;
	}

	@Override
	public boolean waitForAvailable(long timeoutMillis) {

		//只要有一个提供者可用 就返回true
		if (connector.isDirectoryAvailable(directory)) {
			return true;
		}
		long remains = TimeUnit.MILLISECONDS.toNanos(timeoutMillis);

		boolean available = false;
		lock.lock();
		try {
			needSignal.set(true);
			while (!(available = connector.isDirectoryAvailable(directory))) {
				if ((remains = condition.awaitNanos(remains)) <= 0) {
					break;
				}
			}
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			lock.unlock();
		}

		return available || connector.isDirectoryAvailable(directory);

	}
}
